package org.framework.lazy.cloud.network.heartbeat.common.adapter;

import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.permeate.HandleChannelFlowAdvanced;
import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.ChannelProxyFlow;
import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.proxy.HandleChannelProxyFlowAdvanced;
import org.framework.wu.framework.queue.Message;
import org.framework.wu.framework.queue.MessageQueue;
import org.framework.wu.framework.queue.MessageQueueFactory;
import org.springframework.boot.CommandLineRunner;
import org.wu.framework.core.utils.ObjectUtils;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 通道流量适配器
 *
 * @see HandleChannelFlowAdvanced
 */
@Slf4j
public class ChannelProxyFlowAdapter implements CommandLineRunner {


    private static final String QUEUE_NAME = "channel_proxy_flow";

    ThreadPoolExecutor CHANNEL_FLOW_ADAPTER_EXECUTOR =
            new ThreadPoolExecutor(10, 20, 3L, TimeUnit.MINUTES,
                    new LinkedBlockingDeque<>(50),new ThreadPoolExecutor.AbortPolicy());
    // 线程使用完后使用主线程执行

    MessageQueue queue = MessageQueueFactory.getQueue(QUEUE_NAME);

    protected final List<HandleChannelProxyFlowAdvanced> handleChannelProxyFlowAdvancedList;

    public ChannelProxyFlowAdapter(List<HandleChannelProxyFlowAdvanced> handleChannelProxyFlowAdvancedList) {
        this.handleChannelProxyFlowAdvancedList = handleChannelProxyFlowAdvancedList;
    }


    /**
     * 处理当前数据
     *
     * @param channelProxyFlow 通道数据
     */
    public void handler(Channel channel, ChannelProxyFlow channelProxyFlow) {
        for (HandleChannelProxyFlowAdvanced handleChannelProxyFlowAdvanced : handleChannelProxyFlowAdvancedList) {
            if (handleChannelProxyFlowAdvanced.support(channelProxyFlow)) {
                try {
                    handleChannelProxyFlowAdvanced.handler(channel, channelProxyFlow);
                } catch (Exception e) {
                    log.error("流量统计失败:{}", e.getMessage());
                }
                return;
            }
        }
    }

    /**
     * 异步处理当前数据
     *
     * @param channelProxyFlow 通道数据
     */
    public void asyncHandler(Channel channel, ChannelProxyFlow channelProxyFlow) {
     // TODO 流量并发异常
        Message message = new Message("channel_proxy_flow_topic", channelProxyFlow);
        // 消息发送到队列
        queue.send(message);
    }

    @Override
    public void run(String... args) throws Exception {
        // 创建监听线程
        Thread thread = new Thread(() -> {
            while (true) {
                Message receive = queue.receive();
                if(ObjectUtils.isNotEmpty(receive)) {
                    Object body = receive.getBody();
                    handler(null, (ChannelProxyFlow) body);
                }
            }

        });
        CHANNEL_FLOW_ADAPTER_EXECUTOR.submit(thread::start);

    }
}
